package local

import (
	"encoding/json"
	"github.com/sirupsen/logrus"
	"github.com/tluo-github/super-runner/common"
	"github.com/tluo-github/super-runner/helpers/kafka"
	"github.com/tluo-github/super-runner/executors"
	"time"
)
type LocalExecutor struct {
	build       common.Build
	config common.RunnerConfig
}

type KafkaJobLogMsg struct {
	MetaData map[string]interface{}`json:"metadata,omitempty"`
	Event   string		`json:"event"`
	MinioUrl string	   `json:"minio_url"`
	Timestamp   int64  `json:"timestamp"`
}

func (l *LocalExecutor) Prepare(config common.RunnerConfig, build common.Build) error {
	logrus.Info("Prepare")
	l.build = build
	l.config = config
	msg := KafkaJobLogMsg{
		MetaData:  build.MetaData,
		Event:     "start",
		MinioUrl:  "",
		Timestamp: time.Now().Unix(),
	}
	bytes,_ := json.Marshal(&msg)
	kafka.GetKafkaHelper(config.KafkaConfig).WriteBytes(bytes)
	return nil
}

func (l *LocalExecutor) Run() error {
	logrus.Info("Run")
	time.Sleep(time.Second * 1000)
	return nil
}

func (l *LocalExecutor) Wait() error {
	logrus.Info("Wait")
	return nil
}

func (l *LocalExecutor)SendError(err error)  {

}
func (l *LocalExecutor) Cleanup() error {
	logrus.Info("Cleanup")
	msg := KafkaJobLogMsg{
		MetaData:  l.build.MetaData,
		Event:     "succeeded",
		MinioUrl:  "http://xxx",
		Timestamp: time.Now().Unix(),
	}
	bytes,_ := json.Marshal(&msg)
	kafka.GetKafkaHelper(l.config.KafkaConfig).WriteBytes(bytes)
	return nil
}

func createFn() common.Executor {
	return &LocalExecutor{}
}
func init()  {
	common.RegisterExecutor("local",executors.DefaultExecutorProvider{Creator: createFn})
}